/**
 *    Copyright 2015 the original author or authors.
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */
package org.mybatis.caches.redis;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.ReadWriteLock;

import org.apache.ibatis.cache.Cache;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import redis.clients.jedis.DefaultJedisClientConfig;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisClientConfig;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisSharding;

/**
 * Cache adapter for Redis.
 *
 * @author Eduardo Macarron
 */
public final class RedisCache implements Cache {
	private static final Logger LOGGER = LoggerFactory.getLogger(RedisConfigurationBuilder.class);
	private final ReadWriteLock readWriteLock = new DummyReadWriteLock();

	private String id;
	private Integer deployType;
	private static JedisPool pool;
	private static JedisSharding JedisShardingPool;// 切片额客户端连接
	private static JedisCluster cluster;
	private static RedisConfig redisConfig = RedisConfigurationBuilder.getInstance().parseConfiguration();

	public RedisCache(final String id) {
		LOGGER.info("init redis start");
		if (id == null) {
			throw new IllegalArgumentException("Cache instances require an ID");
		}
		LOGGER.info("redisConfig=" + redisConfig);
		this.id = redisConfig.getCache4SystemName() + ":" + id;
		LOGGER.info("id=" + this.id);
		this.deployType = redisConfig.getDeployType();
		switch (redisConfig.getDeployType()) {
		case 0:
			String address = redisConfig.getAddress();
			pool = new JedisPool(redisConfig, address.split(":")[0], Integer.valueOf(address.split(":")[1]),
					redisConfig.getConnectionTimeout());
			break;
		case 1:
			// 池基本配置
			JedisClientConfig config = DefaultJedisClientConfig.builder().build();
			List<HostAndPort> shards = new ArrayList<HostAndPort>();
			// slave链接
			for (String add : redisConfig.getAddress().split(",")) {
				shards.add(new HostAndPort(add.split(":")[0], Integer.valueOf(add.split(":")[1])));
			}
			// 构造池
			JedisShardingPool = new JedisSharding(shards, config);
			break;
		case 2:
			// slave链接
			Set<HostAndPort> jedisClusterNodes = new HashSet<HostAndPort>();
			for (String add : redisConfig.getAddress().split(",")) {
				jedisClusterNodes.add(new HostAndPort(add.split(":")[0], Integer.valueOf(add.split(":")[1])));
			}
			// 构造池
			cluster = new JedisCluster(jedisClusterNodes);
			break;

		default:
			break;
		}
		LOGGER.info("init redis end");
	}

	private Object execute(RedisCallback callback) {
		switch (this.deployType) {
		case 0:
			Jedis jedis = pool.getResource();
			try {
				return callback.doWithRedis(jedis);
			} finally {
				jedis.close();
			}
		case 1:
			try {
				return callback.doWithRedis(JedisShardingPool);
			} finally {
				JedisShardingPool.close();
			}
		case 2:
			return callback.doWithRedis(cluster);
		default:
			Jedis jedis2 = pool.getResource();
			try {
				return callback.doWithRedis(jedis2);
			} finally {
				jedis2.close();
			}
		}
	}

	@Override
	public String getId() {
		return this.id;
	}

	@Override
	public int getSize() {
		return (Integer) this.execute(new RedisCallback() {
			@Override
			public Object doWithRedis(Jedis jedis) {
				Map<byte[], byte[]> result = jedis.hgetAll(RedisCache.this.id.toString().getBytes());
				return result.size();
			}

			@Override
			public Object doWithRedis(JedisSharding jedis) {
				Map<byte[], byte[]> result = jedis.hgetAll(RedisCache.this.id.toString().getBytes());
				return result.size();
			}

			@Override
			public Object doWithRedis(JedisCluster jedis) {
				Map<byte[], byte[]> result = jedis.hgetAll(RedisCache.this.id.toString().getBytes());
				return result.size();
			}
		});
	}

	@Override
	public void putObject(final Object key, final Object value) {
		this.execute(new RedisCallback() {
			@Override
			public Object doWithRedis(Jedis jedis) {
				jedis.hset(RedisCache.this.id.toString().getBytes(), key.toString().getBytes(),
						SerializeUtil.serialize(value));
				jedis.expire(RedisCache.this.id, redisConfig.getKeyMaxLiveMinutes() * 60 * 100);
				return null;
			}

			@Override
			public Object doWithRedis(JedisSharding jedis) {
				jedis.hset(RedisCache.this.id.toString().getBytes(), key.toString().getBytes(),
						SerializeUtil.serialize(value));
				jedis.expire(RedisCache.this.id, redisConfig.getKeyMaxLiveMinutes() * 60 * 100);
				return null;
			}

			@Override
			public Object doWithRedis(JedisCluster jedis) {
				jedis.hset(RedisCache.this.id.toString().getBytes(), key.toString().getBytes(),
						SerializeUtil.serialize(value));
				jedis.expire(RedisCache.this.id, redisConfig.getKeyMaxLiveMinutes() * 60 * 100);
				return null;
			}
		});
	}

	@Override
	public Object getObject(final Object key) {
		return this.execute(new RedisCallback() {
			@Override
			public Object doWithRedis(Jedis jedis) {
				return SerializeUtil
						.unserialize(jedis.hget(RedisCache.this.id.toString().getBytes(), key.toString().getBytes()));
			}

			@Override
			public Object doWithRedis(JedisSharding jedis) {
				return SerializeUtil
						.unserialize(jedis.hget(RedisCache.this.id.toString().getBytes(), key.toString().getBytes()));
			}

			@Override
			public Object doWithRedis(JedisCluster jedis) {
				return SerializeUtil
						.unserialize(jedis.hget(RedisCache.this.id.toString().getBytes(), key.toString().getBytes()));
			}
		});
	}

	@Override
	public Object removeObject(final Object key) {
		return this.execute(new RedisCallback() {
			@Override
			public Object doWithRedis(Jedis jedis) {
				return jedis.hdel(RedisCache.this.id.toString(), key.toString());
			}

			@Override
			public Object doWithRedis(JedisSharding jedis) {
				return jedis.hdel(RedisCache.this.id.toString(), key.toString());
			}

			@Override
			public Object doWithRedis(JedisCluster jedis) {
				return jedis.hdel(RedisCache.this.id.toString().getBytes(), key.toString().getBytes());
			}
		});
	}

	@Override
	public void clear() {
		this.execute(new RedisCallback() {
			@Override
			public Object doWithRedis(Jedis jedis) {
				jedis.del(RedisCache.this.id.toString());
				return null;
			}

			@Override
			public Object doWithRedis(JedisSharding jedis) {
				jedis.del(RedisCache.this.id.toString());
				return null;
			}

			@Override
			public Object doWithRedis(JedisCluster jedis) {
				jedis.del(RedisCache.this.id.toString().getBytes());
				return null;
			}
		});

	}

	@Override
	public ReadWriteLock getReadWriteLock() {
		return this.readWriteLock;
	}

	@Override
	public String toString() {
		return "Redis {" + this.id + "}";
	}

}
